package com.learn.kafka.partition;

import com.learn.kafka.config.ProducerProperties;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.concurrent.ExecutionException;

/**
 * 使用自定义分区器的生产者（继承通用配置父类）
 * @author: learn.kafka
 */
public class ProducerAutoPartition extends ProducerProperties {


    public static void main(String[] args) throws ExecutionException, InterruptedException {
        new ProducerAutoPartition().sendMessages();
    }

    // 构造方法：初始化自定义分区器等特殊配置
    public ProducerAutoPartition() {
        super(); // 调用父类构造方法，加载通用配置
        initCustomConfig();
    }

    // 初始化自定义配置（覆盖父类默认值或添加新配置）
    private void initCustomConfig() {
        // 覆盖父类的 acks 配置（-1 等价于 all）
        setProperty(ProducerConfig.ACKS_CONFIG, "-1");
        // 覆盖 linger.ms 配置
        setProperty(ProducerConfig.LINGER_MS_CONFIG, "10");
        // 覆盖 max.block.ms 配置
        setProperty(ProducerConfig.MAX_BLOCK_MS_CONFIG, "5000");
//        // 设置值序列化器为 IntegerSerializer（父类默认是 StringSerializer）
//        setConfig("value.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
        // 配置自定义分区器
        setProperty(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.learn.kafka.partition.SimplePartitioner");
    }

    // 发送消息的逻辑
    private void sendMessages() throws ExecutionException, InterruptedException {
        // 从父类获取生产者实例（泛型指定为 <String, Integer>）
        Producer<Integer, String> producer = createProducer();

        String topic = "qs4part2673";
        // 获取主题的分区数量
        int partitionSize = producer.partitionsFor(topic).size();
        System.out.println("Partition size: " + partitionSize);

        // 发送10条消息
        for (int i = 0; i < 10; i++) {
            // 创建消息记录（key为i的字符串形式，value为i）
            ProducerRecord<Integer, String> producerRecord = new ProducerRecord<>(topic, i, String.valueOf(i));
            // 同步发送消息（获取元数据）
            RecordMetadata metadata = producer.send(producerRecord).get();
            // 打印发送结果（分区和偏移量）
            System.out.printf("Sent to partition: %d, offset: %d%n", metadata.partition(), metadata.offset());
        }

        // 关闭生产者
        producer.close();
    }

}
